package com.zzjson.zookeeper.client;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.codehaus.jackson.map.ObjectMapper;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * <p>****************************************************************************</p>
 * <li>Description : TODO </li>
 * <li>Version     : 1.0.0</li>
 * <li>Creation    : 2021年03月19日</li>
 * <li>@author     : zzy0_0</li>
 * </ul>
 * <p>****************************************************************************</p>
 */
@Slf4j
public class ConfigCenter implements Watcher {

	private final static String CONNECT_STR = "my-server:2182";

	private final static Integer SESSION_TIMEOUT = 30 * 1000;

	private ZooKeeper zooKeeper = null;

	ObjectMapper objectMapper = new ObjectMapper();

	private final CountDownLatch countDownLatch = new CountDownLatch(1);

	public ConfigCenter() {
		zooKeeper = initConnection();

	}

	@SneakyThrows
	private ZooKeeper initConnection() {
		ZooKeeper zooKeeper = new ZooKeeper(CONNECT_STR, SESSION_TIMEOUT, event -> {
			if (event.getType() == Watcher.Event.EventType.None
					&& event.getState() == Watcher.Event.KeeperState.SyncConnected) {
				log.info("连接已建立");
				countDownLatch.countDown();
			}
		});
		//因为创建连接是daemon线程，一旦业务线程退出，就会退出
		//内部有两个线程 1.发送线程 2.事件监听线程
		countDownLatch.await();
		return zooKeeper;
	}

	@SneakyThrows
	public String createNode(String path, MyConfig myConfig) {
		byte[] bytes = objectMapper.writeValueAsBytes(myConfig);
		//TODO 如果存在就会报错
		return zooKeeper.create(path, bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
	}

	@SneakyThrows
	public static void main(String[] args) {
		ConfigCenter configCenter = new ConfigCenter();

		String path = "/myconfig";

		MyConfig myConfig = new MyConfig();
		myConfig.setKey("anykey");
		myConfig.setName("anyName");

		String node = configCenter.createNode(path, myConfig);


		MyConfig configValue = configCenter.getDataFromZookeeper(path);
		log.info("原始数据: {}", configValue);

		TimeUnit.SECONDS.sleep(Integer.MAX_VALUE);
	}


	@SneakyThrows
	private MyConfig getDataFromZookeeper(String path) {
		byte[] data = zooKeeper.getData(path, this, null);

		MyConfig originalMyConfig = objectMapper.readValue(new String(data), MyConfig.class);
		return originalMyConfig;
	}


	@SneakyThrows
	@Override
	public void process(WatchedEvent event) {
		if (event.getType() == Event.EventType.NodeDataChanged
				&& event.getPath() != null && event.getPath().equals("/myconfig")) {
			log.info(" PATH:{}  发生了数据变化", event.getPath());

			byte[] data = zooKeeper.getData("/myconfig", this, null);

			MyConfig newConfig = objectMapper.readValue(new String(data), MyConfig.class);

			log.info("数据发生变化: {}", newConfig);

		}
	}
}